package com.example.demo.dao;

import com.example.demo.Config.HBaseConfig;
import com.example.demo.pojo.DataPojo;
import com.example.demo.utils.DataProcessUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * HBaseDao 存放各种查询方法
 * @author WZH
 * @create 2021-10-24 20:53
 * @desc
 **/
@Slf4j
@Repository
public class HBaseDao {

    private Connection connection=HBaseConfig.getConnection();
    /**
     * 通过rowke查询某一行的数据
     * @param
     * @return {}
     * @author WZH
     * @date 2021/9/12 0:13
     */
    public  void queryByRowKey(String tableName,String rowKey) throws IOException {
//        1. 获取操作表的table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
//        2. 创建get对象
        Get get = new Get(Bytes.toBytes(rowKey));
//        3. 执行get请求
        Result result = table.get(get);
//        4. 查看result
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            // 打印列蔟名
            System.out.print("行键："+ Bytes.toString(CellUtil.cloneRow(cell))+"  ");
            System.out.print("列簇："+Bytes.toString(CellUtil.cloneFamily(cell))+"  ");
            System.out.print("列："+ Bytes.toString(CellUtil.cloneQualifier(cell))+"  ");
            System.out.println("值："+ Bytes.toString(CellUtil.cloneValue(cell))+"  ");
        }
        table.close();
    }

    /**
     * 获取所有数据
     * @param tableName
     * @return {}
     * @author WZH
     * @date 2021/9/14 12:40
     */
    public  List<DataPojo> getAllRows(String tableName) throws IOException{
        Table hTable = connection.getTable(TableName.valueOf(tableName));
        //得到用于扫描region的对象
        Scan scan = new Scan();
        //使用HTable得到resultcanner实现类的对象
        ResultScanner resultScanner = hTable.getScanner(scan);
        List<DataPojo> dataList = DataProcessUtils.getDataList(resultScanner);
        return dataList;
    }

    /**
     * 通过rowkey 模糊查询某个时间的某个表格的数据
     * @param time :时间字符串 形式：yyyyMMddHH... 2021091913
     * @return {{@link List<DataPojo>}}
     * @author WZH
     * @date 2021/9/19 14:08
     */
    public  List<DataPojo> fuzzyQueryByRowKey(String time,String tableName) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        // 通过正则匹配 rowkey
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(".*" + time + ".*"));
        scan.setFilter(rowFilter);
        ResultScanner scanner = table.getScanner(scan);
        //通过工具类来封装hbase查询到的数据
        List<DataPojo> dataList = DataProcessUtils.getDataList(scanner);
        return dataList;
    }

    /**
     * 通过rowkey 降采样 查询某一时间段的数据
     * @return {{@link List<DataPojo>}}
     * @author WZH
     * @date 2021/9/22 15:21
     */
    public List<DataPojo> randomDownSampleQuery(String startTime, String endTime, String tableName, String sensorId, float SampleRate) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        long time1 =new Date().getTime();

        //设置查询范围
        scan.withStartRow(Bytes.toBytes(sensorId+startTime));
        scan.withStopRow(Bytes.toBytes(sensorId+endTime));


        //        随机过滤，实现降采样 需要传入一个 0-1之间的float型数字.
        RandomRowFilter randomRowFilter = new RandomRowFilter(new Float(SampleRate));
        scan.setFilter(randomRowFilter);

        ResultScanner scanner = table.getScanner(scan);
        List<DataPojo> dataList = DataProcessUtils.getDataList(scanner);

        // 关闭资源
        scanner.close();
        table.close();

        long time2 = new Date().getTime();
        long l = (time2 - time1) / 1000L;
        log.info("本次降查询一共耗时"+l+"秒");
        log.info("一共查询了"+dataList.size()+"记录");
        log.info("一共查询了"+dataList.size()+"条记录");
        return dataList;
    }


    /**
     * 可降采样查询某一时间范围内多个传感器的值
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @param tableName HBASE表名称
     * @param siteList  存放带查询的测点的集合
     * @param sampleRate 降采样频率，在 0-1 之间，若 < 0，则不进行降采样
     * @return {{@link List<DataPojo>}}
     * @author WZH
     * @date 2021/10/20 20:49
     */
    public List<DataPojo> multiSiteQuery(
            String startTime, String endTime,String tableName
            , List<String> siteList
            , float sampleRate
    ) throws IOException {

        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();

        long time1 =new Date().getTime();

        //创建一个 RowRange 的 List 实现多值过滤
        ArrayList<MultiRowRangeFilter.RowRange> rowRanges = new ArrayList<>();
        for (String site : siteList) {
            MultiRowRangeFilter.RowRange rowRange = new MultiRowRangeFilter.RowRange(
                    site+startTime, true,
                    site+endTime, true);
            rowRanges.add(rowRange);
        }
        MultiRowRangeFilter multiRowRangeFilter = new MultiRowRangeFilter(rowRanges);

        //        过滤器组合列表 所有过滤器，与操作
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL,multiRowRangeFilter);



////        随机过滤，实现降采样
        if (sampleRate < 1f && sampleRate >0f) {
            RandomRowFilter randomRowFilter = new RandomRowFilter(sampleRate);
            filterList.addFilter(randomRowFilter);
        }


        scan.setFilter(filterList);
        ResultScanner scanner = table.getScanner(scan);

        List<DataPojo> dataList = DataProcessUtils.getDataList(scanner);
        // 关闭资源
        scanner.close();
        table.close();

        long time2 = new Date().getTime();
        long l = (time2 - time1) / 1000L;
        log.info("本次降查询一共耗时"+l+"秒");
        log.info("一共查询了"+dataList.size()+"记录");
        log.info("本次查询使用了"+filterList.size()+"个过滤器");
        return dataList;
    }

    public static void main(String[] args) {
        HBaseDao hBaseDao =new HBaseDao();
        try {
//            hBaseDao.queryByRowKey("WD_TABLE","WD00020211025111818932");
//            List<DataPojo> dataList = hBaseDao.getAllRows("WD_TABLE");


            //查询2021-10-25 15:03:46的数据
//            List<DataPojo> dataList = hBaseDao.fuzzyQueryByRowKey("20211025150346", "WD_TABLE");

//            List<DataPojo> dataList = hBaseDao.randomDownSampleQuery("202110251503"
////                    , "202110251504"
////                    , "WD_TABLE"
////                    , "WD001"
////                    , 0.5f);

            ArrayList<String> siteList = new ArrayList<>();
            siteList.add("WD000");
            siteList.add("WD002");
            siteList.add("WD003");
            List<DataPojo> dataList = hBaseDao.multiSiteQuery("202110251503"
                    , "202110251504"
                    , "WD_TABLE"
                    , siteList,
                    0.01f
            );
            for (DataPojo dataPojo : dataList) {
                System.out.println(dataPojo);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
